package com.huidu.bitrade.consumer;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.huidu.bitrade.config.MQTraceAspect;
import com.huidu.bitrade.constant.BooleanEnum;
import com.huidu.bitrade.constant.WithdrawStatus;
import com.huidu.bitrade.entity.Coin;
import com.huidu.bitrade.entity.Member;
import com.huidu.bitrade.entity.MemberWallet;
import com.huidu.bitrade.entity.WithdrawRecord;
import com.huidu.bitrade.service.CoinService;
import com.huidu.bitrade.service.MemberService;
import com.huidu.bitrade.service.MemberWalletService;
import com.huidu.bitrade.service.WithdrawRecordService;
import com.huidu.bitrade.util.MessageResult;
import com.huidu.bitrade.vendor.provider.SMSProvider;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.skywalking.apm.toolkit.trace.Trace;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.math.BigDecimal;
import java.util.Collections;
import java.util.Objects;

@Slf4j
@Component
@AllArgsConstructor
public class FinanceConsumer {

    private final CoinService coinService;
    private final MemberWalletService walletService;
    private final MemberService memberService;
    private final RestTemplate restTemplate;
    private final WithdrawRecordService withdrawRecordService;
    private final SMSProvider smsProvider;

    /**
     * 处理充值消息，key值为币种的名称（注意是全称，如Bitcoin）
     *
     * @param record
     */
    @Trace(operationName = "consumerTag-FinanceConsumer-handleDeposit")
    @KafkaListener(topics = {"deposit"})
    public void handleDeposit(ConsumerRecord<String, String> record) {
        log.info("topic={},key={},value={}", record.topic(), record.key(), record.value());
        MQTraceAspect.consumerData(record.topic(),record,"处理充值消息，key值为币种的名称（注意是全称，如Bitcoin）");
        if (StringUtils.isEmpty(record.value())) {
            return;
        }
        JSONObject json = JSON.parseObject(record.value());
        if (json == null) {
            return;
        }
        BigDecimal amount = json.getBigDecimal("amount");
        String txId = json.getString("txid");
        String address = json.getString("address");
        Coin coin = coinService.findOne(record.key());
        if(Objects.isNull(coin)){
            log.warn("处理充值消息没有找到coin:{}",record.key());
            return;
        }
        log.info("coin={}", coin);
        MessageResult messageResult = MessageResult.error(500,"处理充值失败");
        if(coin.getAccountType() == 1) {
        	Long memoId = json.getLong("userId"); // 备注Memo
        	Long userId = memoId - 345678; // 注意，此处与前端memo必须保持一致
        	Member member = memberService.findOne(userId);
        	if(member != null && walletService.findDepositByTxid(txId) == null) {
                messageResult = walletService.recharge2(userId, coin, address, amount, txId);
                log.info("wallet recharge result:{}", messageResult);
        	}
        }else {
	        if (walletService.findDeposit(address, txId) == null) {
                messageResult = walletService.recharge(coin, address, amount, txId);
                log.info("wallet recharge result:{}", messageResult);
	        }
        }
        if(messageResult.getCode()==0){
            MemberWallet wallet = walletService.findByCoinAndAddress(coin, address);
            Member mRes = memberService.findOne(wallet.getMemberId());
            try {
                log.info("========开始发送短信======={}",mRes);
                smsProvider.sendMessageByTempId(mRes.getMobilePhone(), Collections.singletonList("充值" + wallet.getCoin().getUnit() + "成功.充值数量为:" + amount.stripTrailingZeros().toPlainString()),2);
            } catch (Exception e) {
                log.error("========发送短信异常======={}",mRes,e);
            }
        }
    }

    /**
     * 处理提交请求,调用钱包rpc，自动转账
     *
     * @param record
     */
    @Trace(operationName = "consumerTag-FinanceConsumer-handleWithdraw")
    @KafkaListener(topics = {"withdraw"})
    public void handleWithdraw(ConsumerRecord<String, String> record) {
        log.info("topic={},key={},value={}", record.topic(), record.key(), record.value());
        MQTraceAspect.consumerData(record.topic(),record,"处理提交请求,调用钱包rpc，自动转账");
        if (StringUtils.isEmpty(record.value())) {
            return;
        }
        JSONObject json = JSON.parseObject(record.value());
        Long withdrawId = json.getLong("withdrawId");
        try {
            String serviceName = "SERVICE-RPC-" + record.key().toUpperCase();
            String url = "http://" + serviceName + "/rpc/withdraw?address={1}&amount={2}&fee={3}";

            Coin coin = coinService.findByUnit(record.key());
            log.info("coin = {}",coin.toString());
            if (coin != null && coin.getCanAutoWithdraw() == BooleanEnum.IS_TRUE) {
                BigDecimal minerFee = coin.getMinerFee();
                MessageResult result = restTemplate.getForObject(url,
                        MessageResult.class, json.getString("address"), json.getBigDecimal("arriveAmount"), minerFee);
                log.info("=========================rpc 结束================================");
                log.info("result = {}", result);
                if (result.getCode() == 0 && result.getData() != null) {
                    log.info("====================== 处理成功,data为txid更新业务 ==================================");
                    //处理成功,data为txid，更新业务订单
                    String txid = (String) result.getData();
                    withdrawRecordService.withdrawSuccess(withdrawId, txid);
                } else {
                    log.info("====================== 自动转账失败，转为人工处理 ==================================");
                    //自动转账失败，转为人工处理
                    withdrawRecordService.autoWithdrawFail(withdrawId);
                }
            }
        } catch (Exception e) {
            log.error("auto withdraw failed,error={}",e.getMessage(),e );
//            自动转账失败，转为人工处理
            withdrawRecordService.autoWithdrawFail(withdrawId);
        }
    }


    /**
     * 异步打钱后返回状态
     * @param record
     */
    @Trace(operationName = "consumerTag-FinanceConsumer-withdrawNotify")
    @KafkaListener(topics = {"withdraw-notify"})
    public void withdrawNotify(ConsumerRecord<String, String> record){
        log.info("topic={},accessKey={},value={}", record.topic(), record.key(), record.value());
        MQTraceAspect.consumerData(record.topic(),record,"异步打钱后返回状态");
        if (StringUtils.isEmpty(record.value())) {
            return;
        }
        JSONObject json = JSON.parseObject(record.value());
        Long withdrawId = json.getLong("withdrawId");
        WithdrawRecord withdrawRecord=withdrawRecordService.findOne(withdrawId);
        if(withdrawRecord==null){
            return;
        }
        String txid=json.getString("txid");
        int status=json.getInteger("status");
        //转账失败，状态变回等待放币
        if(status==0){
            withdrawRecord.setStatus(WithdrawStatus.WAITING);
            withdrawRecordService.save(withdrawRecord);
        }else if(status==1){
            withdrawRecordService.withdrawSuccess(withdrawId, txid);
        }
    }
}
